package com.zhaosc.rx.example;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.junit.Test;

import io.reactivex.Flowable;
import io.reactivex.Observable;
import io.reactivex.ObservableSource;
import io.reactivex.Scheduler;
import io.reactivex.annotations.NonNull;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.observables.GroupedObservable;
import io.reactivex.parallel.ParallelFlowable;
import io.reactivex.schedulers.Schedulers;

public class ParallelExample {

	@Test
	public void testParallel() {
		Observable.range(1, 100).flatMap(new Function<Integer, ObservableSource<String>>() {
			@Override
			public ObservableSource<String> apply(Integer integer) throws Exception {
				return Observable.just(integer).subscribeOn(Schedulers.computation())
						.map(new Function<Integer, String>() {

							@Override
							public String apply(Integer integer) throws Exception {
								return integer.toString();
							}
						});
			}
		}).subscribe(new Consumer<String>() {
			@Override
			public void accept(String str) throws Exception {

				System.out.println(Thread.currentThread().getName() + "*******" + str);
			}
		});
	}

	@Test
	public void testParallel02() {
		ParallelFlowable<Integer> parallel = Flowable.range(1, 1000).parallel();
		parallel
				// .runOn(Schedulers.io())
				.sequential()
				// .parallel(3, 5)
				.subscribe(new Consumer<Integer>() {

					@Override
					public void accept(Integer t) throws Exception {
						System.out.println(Thread.currentThread().getName() + "********" + t);
					}
				});
	}

	@Test
	public void testParallel03() {
		int threadNum = Runtime.getRuntime().availableProcessors() + 1;

		ExecutorService executor = Executors.newFixedThreadPool(threadNum);
		final Scheduler scheduler = Schedulers.from(executor);
		Observable.range(1, 100).flatMap(new Function<Integer, ObservableSource<String>>() {
			@Override
			public ObservableSource<String> apply(Integer integer) throws Exception {
				return Observable.just(integer).subscribeOn(scheduler).map(new Function<Integer, String>() {

					@Override
					public String apply(Integer integer) throws Exception {
						return integer.toString();
					}
				});
			}
		})
		.doFinally(()->executor.shutdown())
		.subscribe(new Consumer<String>() {
			@Override
			public void accept(String str) throws Exception {

				System.out.println(Thread.currentThread().getName() + "********" + str);
			}
		});
	}
	
	/**
	 * Round-Robin 负载均衡
	 */
	@Test
	public void testParallel04() {
        final AtomicInteger batch = new AtomicInteger(0);

        Observable.range(1,100)
                .groupBy(new Function<Integer, Integer>() {
                    @Override
                    public Integer apply( Integer integer) throws Exception {
                        return batch.getAndIncrement() % 5;
                    }
                })
                .flatMap(new Function<GroupedObservable<Integer, Integer>, ObservableSource<?>>() {
                    @Override
                    public ObservableSource<?> apply( GroupedObservable<Integer, Integer> integerIntegerGroupedObservable) throws Exception {
                        return integerIntegerGroupedObservable.observeOn(Schedulers.io())
                                .map(new Function<Integer, String>() {

                                    @Override
                                    public String apply( Integer integer) throws Exception {
                                        return integer.toString();
                                    }
                                });
                    }
                })
                .subscribe(new Consumer<Object>() {
                    @Override
                    public void accept( Object o) throws Exception {
                    	System.out.println(Thread.currentThread().getName() + "********" + o);
                    }
                });
	}

}
